package comparator;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.StringTokenizer;

import matchers.JaroWinkler;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import file.FileSD;

public class BDM {

	public static IntWritable partitionIndex;
	public static long sumPairs;
	public static ArrayList<BDMData> BDMData = new ArrayList<BDMData>();
	public static int blockingPos = 0;
	public static HashMap<String, Integer> blockingkeyIndexer = new HashMap<String, Integer>();
	
	public static class Map extends Mapper<LongWritable, Text, Text, FileSD> {

		private final static IntWritable one = new IntWritable(1);
		private int numberOfReplications = 5;
		private int token = 5;
		private Text word = new Text();
		private Text k = new Text();

		@Override
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String pathName = ((FileSplit) context.getInputSplit()).getPath().getName();
			int numberOfFile = getNumberOfFile(pathName);

			StringTokenizer itr = new StringTokenizer(value.toString());
			while (itr.hasMoreTokens()) {
				word.set(itr.nextToken());
				if (numberOfFile == 2) {
					for (int i = 0; i < numberOfReplications; i++) {
						k.set(i +""/*+ "." + numberOfFile*/);
						context.write(k, new FileSD(word, new IntWritable(numberOfFile)));
					}
				} else {
					k.set(--token +""/*+ "." + numberOfFile*/);
					context.write(k, new FileSD(word, new IntWritable(numberOfFile)));
				}
				
			}
		}
		// public void map(LongWritable key, Text value, Context context) throws
		// IOException, InterruptedException {
		// word.set(value);
		// String blockingKey = computeKey(word.toString().toUpperCase());
		// word.set(blockingKey + "." +
		// getMapIndex(context.getInputSplit().toString()));
		// context.write(word, one);
		// }

		private int getNumberOfFile(String pathName) {
			return Character.getNumericValue(pathName.charAt(3));
		}

		// private int getMapIndex(String taskAttemptID){
		// taskAttemptID = taskAttemptID.substring(0,
		// taskAttemptID.lastIndexOf("."));
		// return
		// Integer.parseInt(taskAttemptID.substring(taskAttemptID.lastIndexOf("_")+1,
		// taskAttemptID.length()));
		// }
		//
		// public String computeKey(String entity){
		// if (entity.length() < 3) return ">>>";
		// return entity.substring(0, 3).replace(".", ",");
		// }

	}

	public static class Reduce extends
			Reducer<Text, FileSD, Text, Text> {
		private Text result = new Text();
		private Text k = new Text();
		private int log = 1;

		@Override
		public void reduce(Text key, Iterable<FileSD> values,
				Context context) throws IOException, InterruptedException {
//			Text objToCompare = new Text();
//			for (Text val : values) {
//				if (key.charAt(2) == '1'){
//					objToCompare = val;
//					for (Text val2 : values) {
//						//if(key.charAt(2) != '2'){
//							result.set(objToCompare + " x " + val2 + " = " + 1);
//							k.set(log++ + "");
//							context.write(k, result);
//						//}
//					}
//				}
//			}
			
//			Text objToCompare = new Text();
//			if (key.charAt(2) == '1'){
//				for (Text val : values) {
//					objToCompare = val;
//				}
//			}
//			
//			for (Text val : values) {
//				//if(key.toString().charAt(2) != '1'){
//					result.set(objToCompare + " x " + val + " = " + 1);
//					k.set(log++ + "");
//					context.write(k, result);
//				//}
//			}
			
//			String sum = "";
//			for (FileSD val : values) {
//				//context.write(key, val);
//				sum += (val.getName().toString() + val.getDataset().get() + "|");
//			}
//			result.set(sum);
//			k.set(log++ + "");
//			context.write(k, result);
			
			/* ******************************************************/
			
			List<FileSD> filesDataSetSource =  new ArrayList<FileSD>();
			List<FileSD> filesDataSetTarget =  new ArrayList<FileSD>();
			for (FileSD val : values) {
				if (val.getDataset().get() == 1) {
					filesDataSetSource.add(val);
				} else {
					filesDataSetTarget.add(val);
				}
			}
			
//			String saida = "";
//			for (FileSD fileSource : filesDataSetSource) {
//				for (FileSD fileTarget : filesDataSetTarget) {
//					saida += fileSource.getName() + " - " + fileTarget.getName() + " = " + ComputeMatchers(fileSource.getName(), fileTarget.getName()) + "\n";
//				}
//			}
//			result.set(saida);
//			k.set(log++ + "");
//			context.write(k, result);
			
//			for (FileSD fileSource : filesDataSetSource) {
//				result.set(fileSource.getName() /*+ " - " + fileTarget.getName() + " = " + ComputeMatchers(fileSource.getName(), fileTarget.getName())*/);
//				k.set(log++ + "");
//				context.write(k, result);
//				result.clear();
//				k.clear();
//			}
			
			for (FileSD fileT : filesDataSetTarget) {
				result.set(fileT.getName() /*+ " - " + fileTarget.getName() + " = " + ComputeMatchers(fileSource.getName(), fileTarget.getName())*/);
				k.set(log++ + "");
				context.write(k, result);
				result.clear();
				k.clear();
			}
			
//			String saida = "";
//			for (FileSD fileTarget : filesDataSetTarget) {
//				saida += (fileTarget.getName().toString());
//			}
//			result.set(saida);
//			k.set(log++ + "");
//			context.write(k, result);
//			result.clear();
//			k.clear();
			
//			if (filesDataSetSource.isEmpty() || filesDataSetTarget.isEmpty()) {
//				k.set(log++ + "");
//				result.set("No results");
//				context.write(k, result);
//			}
			/* ******************************************************/
			
//			if (fileDataSetSource != null) {
//				for (FileSD val : values) {
//					//if (val.getDataset().get() == 2) {
//						result.set(fileDataSetSource.getName() + " - " + val.getName() + " = " + ComputeMatchers(fileDataSetSource.getName(), val.getName()));
//						k.set(log++ + "");
//						context.write(k, result);
//					//}
//				}
//			} else {
//				k.set(log++ + "");
//				result.set("No results");
//				context.write(k, result);
//			}
			
			
//			String sum = "";
//			for (Text val : values) {
//				//context.write(key, val);
//				sum += val.toString();
//			}
//			result.set(sum);
//			k.set(key.charAt(0)+"");
//			context.write(k, result);
			
//			Text objToCompare = new Text();
//			for (Text val : values) {
//				if(key.toString().charAt(2) == '1'){
//					objToCompare = val;
//				}
//			}
//			
//			for (Text val : values) {
//				if(key.toString().charAt(2) != '1'){
//					result.set(objToCompare + " x " + val + " = " + ComputeMatchers(objToCompare, val));
//					context.write(key, result);
//				}
//			}
//			//result.set(sum);
			
		}

		private FileSD getDataSetSource(Iterable<FileSD> values) {
			for (FileSD val : values) {
				if (val.getDataset().get() == 1) {
					return val;
				}
			}
			return null;
		}

		private String ComputeMatchers(Text text1, Text text2) {
			JaroWinkler winkler = new JaroWinkler();
			double result = winkler.compare(text1.toString(), text2.toString());
			return new DecimalFormat("#.##").format(result);
		}

		// public void reduce(Text key, Iterable<IntWritable> values, Context
		// context)
		// throws IOException, InterruptedException {
		// int sum = 0;
		// for (IntWritable val : values) {
		// sum += val.get();
		// }
		// context.write(key, new IntWritable(sum));
		// }
	}

	// public static class Map_configure extends Configuration{
	//
	// public void map_configure(Map m, Reduce r, IntWritable partIndex) throws
	// IOException, InterruptedException {
	// //Store
	// partitionIndex = partIndex;
	// }
	// }
	//
	// public static long pairs(){
	// return BDM.sumPairs;
	// }
	//
	// public static ArrayList<BDMData> readBDM(){
	// return BDMData;
	// }
	//
	// public static int numBlocks(){
	// return BDMData.size();
	// }
	//
	// public static int size(int k){
	// return BDMData.get(k).getSize();
	// }
	//
	// public static int size(int k, int i){
	// //return BDMData.get(k).getSize();
	// Object partitionSize = BDMData.get(k).getPartitions().get(i);
	// if (partitionSize != null) return (Integer) partitionSize;
	// return 0;
	// }
	//
	// public static String blockKey(int blockIndex){
	// return BDMData.get(blockIndex).getKey();
	// }
	//
	// public static int blockIndex(String blockingkey){
	// return blockingkeyIndexer.get(blockingkey);
	// }
	//
	// public static int getTotalToPatition(String blockingkey, int partition){
	// HashMap<Integer, Integer> partitions =
	// BDMData.get(blockIndex(blockingkey)).getPartitions();
	// Object[] partitionsSet = partitions.keySet().toArray();
	// int entitiesCount = 0;
	// for (Object object : partitionsSet) {
	// if ((Integer)object < partition){
	// entitiesCount += partitions.get(object);
	// }
	// }
	// return entitiesCount;
	// }
	
//	public class Partition extends Partitioner<Text, Text> {
//
//	    @Override
//	    public int getPartition(Text key, Text value, int numReduceTasks) {
//	        return (key.toString().charAt(0)) % numReduceTasks;
//	    }    
//	}
}